package middleware.kafka.Serializer;

import org.apache.kafka.common.errors.SerializationException;
import org.apache.kafka.common.serialization.Serializer;

import java.io.UnsupportedEncodingException;
import java.util.Map;

/**
 * @Author suxiaoxiao
 * @Date 2019/6/26 22:42
 * @Version 1.0
 */
public class StringSerializer implements Serializer<String> {

    private String encoding = "UTF8";

    /**
     * 创建 KafkaProducer 实例的时候调用，
     * 主要用来确定编码类型。
     * @param configs
     * @param isKey
     */
    @Override
    public void configure(Map<String, ?> configs, boolean isKey) {
        String propertyName = isKey ? "key.serializer.encoding" : "value.serializer.encoding";
        Object encodingValue = configs.get(propertyName);
        if (encodingValue == null) {
            encodingValue = configs.get("serializer.encoding");
        }
        if (encodingValue != null && encodingValue instanceof String) {
            encoding = (String)encodingValue;
        }
    }

    /**
     * 将 String 转换为 byte[]
     * @param topic
     * @param data
     * @return
     */
    @Override
    public byte[] serialize(String topic, String data) {
        try {
            if (data == null) {
                return null;
            } else {
                return data.getBytes(encoding);
            }
        } catch (UnsupportedEncodingException e) {
            throw new SerializationException("Error when serializing " +
                    "string to byte[] due to unsupported encoding " + encoding);
        }
    }

    @Override
    public void close() {
        // nothing to do
    }

}
